package com.ly.kafka.utils;


import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

//消费者API
public class ConsumerDemo {

    public static void main(String[] args) {
        Properties properties = new Properties();
        //集群方式连接以都好隔开
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.23.102:9092,192.168.23.103:9092,192.168.23.105:9092");

        //消费组名
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"testGroup");

        //是否自动提交offset 性能高 会涉及到消息重复
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
        //自动提交offset时间间隔
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
//        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

        /*
         *心跳时间 服务端broker通过心跳确认consumer是否故障，如果发现故障，就会心跳下发rebalance的指令
         * 给其他的consumer通知他们进行rebalance操作，这个时间可以设置的短一点
         */
        properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,1000);
        //服务端broker多久感知不到一个consumer心跳就认为他故障了 默认是10S
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10*1000);

        /*
        * 如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太差，
        * 就会将这个consumer踢出消费组，将分区分配给别的consumer
        * */
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,30*1000);

        //序列化将key和value序列化成字符数组
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);
        //消费主题
        String topicName="hello-kraft1";
        /****************消费几种模式start 根据需求任选一组就行*************************/
        //如果不主动设置分区 就会轮询的去每个分区都会消费
        consumer.subscribe(Arrays.asList(topicName));

        //也可以指定分区消费 指定了0这个分区
        //consumer.assign(Arrays.asList(new TopicPartition(topicName,0)));

        //消息回溯(以前发送过来的消息)消费
        //consumer.assign(Arrays.asList(new TopicPartition(topicName,0)));
        //consumer.seekToBeginning(Arrays.asList(new TopicPartition(topicName,0)));

        //指定offset消费 不是从0开始了 每次都是从offset=10开始消费信息
        //consumer.assign(Arrays.asList(new TopicPartition(topicName,0)));
        //consumer.seek(new TopicPartition(topicName,0),10);

        //指定时间段去消费 这个应用场景 可用于查看某个时间段消费的信息 用于调试昨天日志消息 可能不对 看看昨天的消费数据
        // 原理就是根据时间算出对应的offset集合 基于offset再去取消费数据。
        /*Map<TopicPartition,Long> map = new HashMap<>();
        List<PartitionInfo> topicPartitions = consumer.partitionsFor(topicName);
        //例如指定半小时前消费
        long startTime = new Date().getTime()- 1000*60*60;
        for (PartitionInfo topicPartition : topicPartitions) {
            //将主题内所有分区数据取出来
            map.put(new TopicPartition(topicName,topicPartition.partition()),startTime);
        }
        //取出分区内所有的offset
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            //分区
            TopicPartition key = entry.getKey();
            //对应的offset
            OffsetAndTimestamp value = entry.getValue();
            if(key==null || value ==null){
                continue;
            }
            //取出offset
            Long offset = value.offset();
            //根据offset取出消费的信息
            if(value!=null){
                consumer.assign(Arrays.asList(key));
                consumer.seek(key,offset);
            }
        }*/
        /****************消费几种模式end*************************/
        //消费代码体
        while (true){
            /*
            * poll() API 是拉取消息的长轮询，主要是判断consumer是否还或者，只要我们持续调用poll()
            * 消费者就会存活在自己所在的group中，并且持续的消费指定partition的消息
            * 底层原理：消费者向server持续发送心跳，如果一个时间段(session.timeout.ms.ms) consumer挂掉
            * 或者不能发送心跳，这个消费者会被认为挂掉了，这个partition也会重新分配给其他的consumer
            * */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到消息：offset = %d，key = %s，value = %s%n",record.offset(),
                        record.key(),record.value());
            }
            if(records.count()>0){
                //提交offset
                consumer.commitSync();
            }
        }
    }
}
